/**
 * class: Node
 * 
 * CharmGil Hong
 */

package centralized;

import java.util.*;
import java.util.concurrent.ConcurrentLinkedQueue;

import common.*;

public class Node extends Thread{
	private int clkspd_ms;
	private long localtime_s;
	private int drift;
	private double smoothing = 1.0;
	
	private long ip;
	private int pid;
	
	private Middleware network;
	
	private int counter;
	private int synced_before_s = 0;
	private int logical_counter;
	
	public ConcurrentLinkedQueue<Message> msg_q;	// a message box (public)
	private LinkedList<Object> holding_q;			// local queue (private)

	private int master_pid = -1;
	private long master_ip = -1;
	private long master_since_logicalcounter = -1;
	private long election_giveup_localtime_s = 0;
	
	private boolean election_raised = true;
	private boolean master_raised = false;
	private boolean init_raised = true;
	private boolean sleep_raised = false;
	private int sleep_left_s;
	private int sleep_counter;
	
	
	public Node( Middleware newNetwork ){
		// hw
		clkspd_ms = Config.TICK_NORMAL_MS;
		drift = ( new Random() ).nextInt( 3 * clkspd_ms / 2 ) - clkspd_ms / 2;	// drift = [ -clkspd_ms/2, 2*clkspd_ms ] 
		// sw
		counter = Config.MAX_COUNTER;
		pid = ( new Random() ).nextInt( Config.MAX_PID );
		logical_counter = 0;
		
		// msg q
		msg_q = new ConcurrentLinkedQueue<Message>();
		// auxiliary q
		holding_q = new LinkedList<Object>();
		
		// network
		network = newNetwork;
		ip = network.connect( pid, this );
		
		print_log( "instantiated and joined to the network." );
	}
	
	
	/**
	 * Run() - Thread body.
	 * 		It periodically generates ticks and update the node timer.
	 * 		It also handles messages received and will be sent.
	 */
	public void run(){
		while( true ){
			
			// clock (tick)
			try{
				// for preemptive sleeping
				if( sleep_raised ){
					sleep( clkspd_ms );		// don't account drift
					
					if( --sleep_counter == 0 ){
						sleep_counter = Config.MAX_COUNTER;
						print_log( "sleeping" );
						
						if( --sleep_left_s == 0 ){
							sleep_raised = false;
							init_raised = true;
							
							smoothing = 1.0;
							master_pid = -1;
							master_ip = -1;
							print_log( "... and I'm back!" );
						}
					}
					
					msg_q.poll();	// emptying msg_q: ignore all incoming messages while sleeping
					
					continue;
				}
				
				sleep( ( int )( ( clkspd_ms + drift ) * smoothing ) );
			}catch( InterruptedException e ){
				e.printStackTrace();
			}
			
			
			// value init
			if( init_raised ){
				if( --counter == 0 ){
					// timer interrupt
					localtime_s++;
					synced_before_s++;
					
					// counter reset
					counter = Config.MAX_COUNTER;
					
					// sync request
					send( Def.SYNC_REQ, 0, 0L, null, logical_counter );
					print_log( "SYNC_REQ" );
					
					logical_counter++;
				}	
				
				Message incoming_msg = msg_q.poll();
				if( incoming_msg != null ){
					
					// logical counter
					logical_counter = Math.max( logical_counter + 1, incoming_msg.getMid() + 1 );
					
					// sync
					if( incoming_msg.getHeader() == Def.ACK && incoming_msg.getContent() != null ){
						if( incoming_msg.getContent().substring( 0, 4 ).equals( "sync" ) ){
							// physical clock of a node is initially 0 -- smoothing is not required!
							localtime_s = Long.parseLong( incoming_msg.getContent().substring( 4 ) );
							master_pid = incoming_msg.getSpid();
							master_ip = incoming_msg.getSip();
							master_since_logicalcounter = logical_counter;
							init_raised = false;
							
							synced_before_s = 0;
							
							print_log( "values are initiated." );
						}
					}
				}
				
				//logical_counter++;	-- it's already increased (above)
				
				if( init_raised )
					continue;
			}
			
			
			// when it becomes a master
			if( master_raised ){
				send( Def.COORDINATOR, 0, 0L, null, logical_counter );
				// receive stuff from the last master (if possible)
				
				
				if( master_pid != pid || master_ip != ip ){	// only if the master actually changed
					print_log( "I am the new sword master! (d=" + drift + ")" );

					// inauguration
					master_pid = pid;
					master_ip = ip;
					master_since_logicalcounter = logical_counter;
					network.set_master( ip, pid );
				}
				
				// cancel my other elections
				for( int sz = holding_q.size(); sz > 0; sz-- ){
					int i = sz - 1;
					Object obj = holding_q.get( i );
					if( obj instanceof ElectionRecord ){	// elections being held by me
						holding_q.remove( obj );
					}
				}
				
				master_raised = false;
				
				logical_counter++;
				
				if( master_raised )
					continue;
			}
			
			
			// timer
			if( --counter == 0 ){
				
				// timer interrupt
				localtime_s++;
				synced_before_s++;
				
				// reset smoothing factor (for the clock adjustment)
				if( synced_before_s >= Config.RESYNC_PERIOD_CENT_S / ( double )Config.SMOOTHING_RATE && smoothing != 1.0 ){
					smoothing = 1.0;
					synced_before_s = 0;	// assumes synchronization is complete at time Config.RESYNC_PERIOD_CENT_S / ( double )Config.SMOOTHING_RATE 
				}else if( smoothing == Config.MAX_COUNTER * ( Config.RESYNC_PERIOD_CENT_S - 1 ) / ( double )( clkspd_ms + drift ) ){
					smoothing = 1.0;
				}
				
				
				// counter reset
				counter = Config.MAX_COUNTER;
				
				// resync
				if( this.is_master() && synced_before_s >= Config.RESYNC_PERIOD_CENT_S ){
					//send RESYNC
					int n_br = send( Def.RESYNC, 0, 0L, "" + localtime_s, logical_counter );
					print_log( "has broadcasted a recync signal to " + n_br + " nodes. (master d:" + drift + ")" );
					
					// timer reset
					synced_before_s = 0;
					
				}else if( !this.is_master() && synced_before_s >= Config.RESYNC_PERIOD_CENT_S * Config.RESYNC_TOLERANCE ){
					// if one misses more than 3 resync signal
					if( localtime_s - election_giveup_localtime_s > Config.ELECTION_GIVEUP_EXPIRE_S ){
						election_raised = true;	// raise election
						for( int sz = holding_q.size(); sz > 0; sz-- ){	// prevent multiple ELECTIONs
							int i = sz - 1;
							Object obj = holding_q.get( i );
							if( obj instanceof ElectionRecord )
								election_raised = false;
						}
					}
				}
				
				logical_counter++;
			}	
			
			
			// election
			if( election_raised && logical_counter - master_since_logicalcounter > Config.MIN_ELECTION_PERIOD_LOGICAL
					&& localtime_s - election_giveup_localtime_s > Config.ELECTION_GIVEUP_EXPIRE_S ){	// if this node indicates an election is needed
				
				boolean ongoing = false;
				
				election_raised = false;
				
				// don't hold ELECTION if I am holding
				for( int sz = holding_q.size(); sz > 0; sz-- ){
					int i = sz - 1;
					Object obj = holding_q.get( i );
					if( obj instanceof ElectionRecord ){	// elections being held by me
						ongoing = true;
					}
				}
				
				if( !ongoing ){
					// hold an election
					ElectionRecord er = new ElectionRecord();
					er.election_time_logical = logical_counter;
					er.election_time_s = localtime_s;
					er.n_participant = send( Def.ELECTION, 0, 0L, "" + er.election_time_s, logical_counter );	// hold an election
					
					print_log( "ELECTION is sent to " + er.n_participant + " nodes." );
					
					holding_q.offer( er );
					
					logical_counter++;					
				}
			}
			
			
			// for my election being held -- checking timeout
			for( int sz = holding_q.size(); sz > 0; sz-- ){
				int i = sz - 1;
				Object obj = holding_q.get( i );
				if( obj instanceof ElectionRecord ){	// elections being held by me
					ElectionRecord er = ( ElectionRecord )obj;
					
					// timeout --> this node will be the next master
					if( localtime_s - er.election_time_s > Config.ELECTION_TIMEOUT_S ){
						print_log( "(election timeout)" );
						master_raised = true;
						
						holding_q.remove( obj );
						logical_counter++;
					}
				}
			}
			
			// msg handle
			Message incoming_msg = msg_q.poll();
			if( incoming_msg != null ){
				// logical counter
				logical_counter = Math.max( logical_counter + 1, incoming_msg.getMid() + 1 );
				
				// msg proc
				switch( incoming_msg.getHeader() ){
					case Def.SYNC_REQ:
						if( is_master() )
							send( Def.ACK, incoming_msg.getSpid(), incoming_msg.getSip(), "sync" + localtime_s, logical_counter );
						
						break;
						
					case Def.RESYNC:
						// smoothing
						// we assumed all nodes know the RESYNC period, since any node can be a master.
						// assumed that the receiver will take the RESYNC signal as UTC
						
						long diff = localtime_s - Long.parseLong( incoming_msg.getContent() );
						
						if( diff >= Config.RESYNC_PERIOD_CENT_S / ( double )Config.SMOOTHING_RATE ){
							// if diff is larger than sync period_s, put the process on hold for Config.RESYNC_PERIOD_CENT_S - 1 sec
							smoothing = Config.MAX_COUNTER * ( Config.RESYNC_PERIOD_CENT_S - 1 ) / ( double )( clkspd_ms + drift );
						}else{
							smoothing = 1.0 - diff * Config.SMOOTHING_RATE / ( double )Config.RESYNC_PERIOD_CENT_S;
							smoothing /= ( 1.0 + diff / ( double )synced_before_s );
							smoothing = 1 / smoothing;
							smoothing = ( smoothing <= 0 ) ? Config.RESYNC_PERIOD_CENT_S / ( double )( clkspd_ms + drift ) : smoothing;	// prevent backward adjustment
							if( smoothing == Double.POSITIVE_INFINITY )	smoothing = 1.0;
							if( smoothing == Double.NaN )	smoothing = 1.0;
							
							if( -diff >= Config.RESYNC_PERIOD_CENT_S / ( double )Config.SMOOTHING_RATE ){
								double extra = -diff / ( Config.SMOOTHING_RATE / ( double )Config.RESYNC_PERIOD_CENT_S );
								smoothing = smoothing / extra; 
							}
								
							//smoothing = ( smoothing == Double.POSITIVE_INFINITY ) ? 1.0 : smoothing;	// prevent permanent sleep
							//smoothing = ( smoothing == Double.NaN ) ? 1.0 : smoothing;	// prevent permanent sleep
						}
						
						print_log( "\tdrft=" + String.format( "%3d", drift ) + "    diff=" + String.format( "%3d", diff )
								//+ "\t s-d=" + df2.format( ( synced_before_s - diff ) )
								+ "   =>   smoothing= " + String.format( "%3.2f", smoothing )
								+ "\ttick= "  + String.format( "%3.2f", ( clkspd_ms + drift ) * smoothing )
								+ "\tsize(msg_q):" + msg_q.size()
								+ "\tsize(holding_q):" + holding_q.size() );
						
						synced_before_s = 0;
						
						send( Def.ACK, incoming_msg.getSpid(), incoming_msg.getSip(), "resc" + incoming_msg.getMid(), logical_counter );
						break;
						
					// the bully algorithm
					case Def.ELECTION:
						// response
						send( Def.ACK, incoming_msg.getSpid(), incoming_msg.getSip(), "elec" + incoming_msg.getMid(), logical_counter );

						if( incoming_msg.getMid() < master_since_logicalcounter ){
							//send( Def.NAK, incoming_msg.getSpid(), incoming_msg.getSip(), "elec" + incoming_msg.getMid(), logical_counter );
							// old packet - ignore
							break;
						}//else if( logical_counter - incoming_msg.getMid() > Config.ELECTION_TIMEOUT_S ){
						//	send( Def.NAK, incoming_msg.getSpid(), incoming_msg.getSip(), "elec" + incoming_msg.getMid(), logical_counter );
						//	break;
						//}
						
						// ELECTION by me
						election_raised = false;
						
						// if incoming ELECTION is higher + I have own ELECTION: I give up
						if( incoming_msg.getSpid() > pid || ( incoming_msg.getSpid() == pid && incoming_msg.getSip() > ip ) ){
							 
							// drop my ELECTION
							for( int sz = holding_q.size(); sz > 0; sz-- ){
								int i = sz - 1;
								Object obj = holding_q.get( i );
								if( obj instanceof ElectionRecord )	// elections being held by me
									holding_q.remove( obj );
							}
							// don't init election for a while
							election_raised = false;
							election_giveup_localtime_s = localtime_s;
							
						}else if( incoming_msg.getSpid() == pid && incoming_msg.getSip() == ip ){
							// ignore my packet
							
						}else{	// if incoming ELECTION is lower than me
							// if self has higher pid (tie breaking is done by ip (no overlap))
							if( incoming_msg.getSpid() < pid || ( incoming_msg.getSpid() == pid && incoming_msg.getSip() < ip ) ){
								if( localtime_s - election_giveup_localtime_s > Config.ELECTION_GIVEUP_EXPIRE_S ){
									election_raised = true;
									for( int sz = holding_q.size(); sz > 0; sz-- ){	// prevent multiple ELECTIONs
										int i = sz - 1;
										Object obj = holding_q.get( i );
										if( obj instanceof ElectionRecord )
											election_raised = false;
									}
								}
							}
							//else
								//master_ltime = logical_counter;
						}
						break;
						
					case Def.ACK:
						// election
						if( incoming_msg.getContent().substring( 0, 4 ).equals( "elec" ) && holding_q.size() > 0 ){
							// compare pid, ip
							if( incoming_msg.getSpid() > pid || ( incoming_msg.getSpid() == pid && incoming_msg.getSip() > ip ) ){
								for( int sz = holding_q.size(); sz > 0; sz-- ){
									int i = sz - 1;
									Object obj = holding_q.get( i );
									if( obj instanceof ElectionRecord ){	// elections being held by me
										// cancel election
										holding_q.remove( obj );
									}
								}
								election_raised = false;
								election_giveup_localtime_s = localtime_s;
							}
						}
						break;
					
					case Def.COORDINATOR:
						if( master_since_logicalcounter < incoming_msg.getMid() ){
							
							master_pid = incoming_msg.getSpid();
							master_ip = incoming_msg.getSip();
							master_since_logicalcounter = logical_counter;
							
							// cancel my elections
							if( incoming_msg.getSpid() > pid || ( incoming_msg.getSpid() == pid && incoming_msg.getSip() > ip ) ){
								for( int sz = holding_q.size(); sz > 0; sz-- ){	// prevent multiple ELECTIONs
									int i = sz - 1;
									Object obj = holding_q.get( i );
									if( obj instanceof ElectionRecord ){	// elections being held by me
										// cancel election
										holding_q.remove( obj );
									}
								}
							}
							
							if( master_pid < pid || ( master_pid == pid && master_ip < ip ) ){
								print_log( "well..." );
								send( Def.NAK, master_pid, master_ip, "coor", logical_counter );
							}else{
								election_raised = false;
								election_giveup_localtime_s = 0; 
								send( Def.ACK, master_pid, master_ip, "coor", logical_counter );
							}
							
						}else if( incoming_msg.getSpid() == master_pid && incoming_msg.getSip() == master_ip ){	// skip
							election_giveup_localtime_s = 0; 
							//send( Def.ACK, master_pid, master_ip, "coor", logical_counter );
						}else{
							send( Def.NAK, master_pid, master_ip, "coor", logical_counter );
							print_log( "New COORDINATOR(" + incoming_msg.getSpid() + "): Rejected. :: " + master_since_logicalcounter + " < " + incoming_msg.getMid() );
						}
						
						break;
						
					case Def.NAK:
						if( incoming_msg.getContent().substring( 0, 4 ).equals( "coor" ) && master_ip == ip ){
							if( incoming_msg.getSpid() < pid || ( incoming_msg.getSpid() == pid && incoming_msg.getRip() < ip ) )
								election_raised = true;
							else
								send( Def.COORDINATOR, incoming_msg.getSpid(), incoming_msg.getSip(), "" + incoming_msg.getMid(), logical_counter );
						}
							
						break;
						
					case Def.ROOT_KILL:
						network.disconnect( ip, pid );
						print_log( "I hate you guys! (Terminated)" );
						try{
							this.stop();
						}catch( Exception e ){}
						// end-of-thread: no more execution
						break;
					
					case Def.ROOT_SLEEP:
						sleep_left_s = Config.ROOT_SLEEP_TIMEOUT_S;
						sleep_counter = Config.MAX_COUNTER;
						sleep_raised = true;

						break;
						
					default:
						print_log( "received an unkown msg from " + incoming_msg.getSpid()  + "|"
								+ Utils.ipToString( incoming_msg.getRip() ) + ".", Def.STDERR );
				}
			}
		}//end-of-switch() -- msg handler
	}//end-of-run()
	
	public int send( int msg_header, int rcv_pid, Long rcv_ip, String appx, int msg_counter ){
		//print_log( "**msg** " + msg_header );
		
		Message outgoing_msg = new Message( msg_counter, msg_header, rcv_pid, rcv_ip, pid, ip );
		if( appx != null ){
			outgoing_msg.setContent( appx );
		}
		
		// update master info
		if( msg_header == Def.COORDINATOR ){
			network.set_master( ip, pid );
		}
		
		// send msg
		if( rcv_pid == 0 ){	// 0 to broadcast
			Collection<Node> all_nodes = network.routing_table.values();
			Iterator<Node> itr = all_nodes.iterator();
			int br_count = 0;	// #msg broadcasted
			while( itr.hasNext() ){
				Node br_rcv = itr.next();	// br_rcv = broadcast, receiver
				
				if( msg_header == Def.ELECTION ){	// for ELECTION, broadcast msg only to higher pid or ip
					if( br_rcv.get_pid() > pid || ( br_rcv.get_pid() == pid && br_rcv.get_ip() > ip ) ){
						br_rcv.msg_q.offer( outgoing_msg );
						br_count++;
					}
				}else{	// normally, broadcast msg to all
					if( br_rcv.get_pid() != pid || ( br_rcv.get_pid() == pid && br_rcv.get_ip() != ip ) ){
						br_rcv.msg_q.offer( outgoing_msg );
						br_count++;
					}
				}
			}
			return br_count;
			
		}else{	// 1-to-1 comm
			Node receiver = network.routing_table.get( new NodeRecord( rcv_pid, rcv_ip ) );
			if( receiver == null )
				return 0;
			
			receiver.msg_q.offer( outgoing_msg );
			return 1;
		}
		
	}//end-of-send()
	
	public void recv(){
		
	}
	
	public boolean is_master(){
		if( master_pid == pid && master_ip == ip )
			return true;
		else
			return false;
	}
	
	public long get_ip(){
		return ip;
	}
	
	public int get_pid(){
		return pid;
	}
	
	public long get_localtime_s(){
		return localtime_s;
	}
	
	public long get_localtime_ms(){
		return localtime_s * 1000;
	}
	
	public int get_drift(){
		return drift;
	}
	
	public void set_localtime_ms( long new_localtime_ms ){
		localtime_s = new_localtime_ms / 1000;
	}
	
	public void set_localtime_s( long new_localtime_s ){
		localtime_s = new_localtime_s;
	}
	
	public void set_master(){
		master_pid = pid;
		master_ip = ip;
		master_since_logicalcounter = logical_counter;
		init_raised = false;
	}
	
	public void set_master( int new_master_pid, long new_master_ip ){
		master_pid = new_master_pid;
		master_ip = new_master_ip;
		master_since_logicalcounter = logical_counter;
		init_raised = false;
	}
		
	public void print_log( String log ){
		print_log( log, Def.STDOUT );
	}
	
	public void print_log( String log, byte target ){
		if( !Config.VERBOSE )
			return;
		
		if( target == Def.STDOUT ){
			System.out.print( "[MSG|" + pid + ":" + Utils.ipToString( ip ) + "(" + localtime_s + ")] " );
			System.out.println( log );
		}else if( target == Def.STDERR ){
			System.err.print( "[ERR|" + pid + ":" + Utils.ipToString( ip ) + "(" + localtime_s + ")] " );
			System.err.println( log );
		}
	}
	
	public String toString(){
		return "Node(" + pid + ":" + Utils.ipToString( ip ) + ")";
	}
	
	class ElectionRecord{
		int election_time_logical;
		long election_time_s;
		int n_participant;
	}
}
